package com.framework.flink;

import cn.hutool.core.util.StrUtil;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.io.RowCsvInputFormat;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.core.fs.Path;
import org.apache.flink.types.Row;

import java.io.*;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

/**
 * 描述：
 *
 *  测试文件写入
 * @author xianggj
 * @Date 2021/10/19 14:58
 **/
public class FlinkTestFileSink {

    public static void main(String[] args) throws Exception{
        File file = new File("D:\\temp\\csvread\\14dbe980254f49d08fd93d4c93ca9dc1.csv");
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        String encoding = "UTF-8";
        Boolean isFirstHeader = false;
        boolean isHeader = isFirstHeader == null? false: isFirstHeader;
        TypeInformation[] fieldTypeInfos = new TypeInformation[16];
        Arrays.fill(fieldTypeInfos, BasicTypeInfo.STRING_TYPE_INFO);
        RowCsvInputFormat rowCsvInputFormat =
                new RowCsvInputFormat(new Path(file.toURI()), fieldTypeInfos);
        //设置编码
        rowCsvInputFormat.setCharset(encoding);
        //跳过第一行
        rowCsvInputFormat.setSkipFirstLineAsHeader(isHeader);
        rowCsvInputFormat.setLenient(true);
        //启用引用字符串解析 当字符串中间有分隔符逗号“,”使用引用做特征字符串转义
        rowCsvInputFormat.enableQuotedStringParsing('\"');
        RowTypeInfo rowTypeInfo = new RowTypeInfo(fieldTypeInfos);
        DataSource<Row> input = env.createInput(rowCsvInputFormat, rowTypeInfo);
        ddd(input);

    }

    private synchronized static void ddd(DataSource<Row> input) throws Exception {
        List<Row> collect = input.collect();
        System.out.println(collect.size());
    }


    /**
     * csv获取文件头
     */
    public static List<String> getHeader(InputStream inputStream, String encoding, boolean isHeader) {
        List<String> header = new ArrayList<>();
        try (BufferedReader brTest = new BufferedReader(new InputStreamReader(inputStream, Charset.forName(encoding)))){
            String text = brTest.readLine();
            String[] split = StrUtil.split(StrUtil.removeAll(text, "\""), ",");
            if (isHeader) {
                header = Arrays.asList(split);
            } else {
                for (int i = 0; i < split.length; i++) {
                    header.add("cc_" + (i+1));
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        return header;
    }
}
